package magnolify.tools;

import joptsimple.OptionParser;
import joptsimple.OptionSet;
import joptsimple.OptionSpec;
import org.apache.avro.Schema;
import org.apache.avro.file.DataFileStream;
import org.apache.avro.generic.GenericDatumReader;
import org.apache.avro.generic.GenericRecord;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.Path;

import java.io.InputStream;
import java.io.PrintStream;
import java.util.List;

public class MagnolifyAvroTool implements Tool {
  @Override
  public String getName() {
    return "avro";
  }

  @Override
  public String getShortDescription() {
    return "Generate Magnolify compatible Scala types from a Avro file";
  }

  @Override
  public int run(InputStream in, PrintStream out, PrintStream err, List<String> args) throws Exception {
    OptionParser optionParser = new OptionParser();
    OptionSpec<Integer> widthOption = optionParser
        .accepts("width", "Column width")
        .withOptionalArg()
        .ofType(Integer.class)
        .defaultsTo(80);
    OptionSpec<Void> sourceOption = optionParser
        .accepts("source", "Include source schema.");

    OptionSet optionSet = optionParser.parse(args.toArray(new String[0]));
    int width = optionSet.valueOf(widthOption);
    boolean source = optionSet.has(sourceOption);
    @SuppressWarnings("unchecked")
    List<String> argz = (List<String>) optionSet.nonOptionArguments();
    if (argz.size() != 1) {
      err.println("avro --width <width> --source input-file");
      err.println();
      err.println(getShortDescription());
      return 1;
    }

    Path path = new Path(argz.get(0));
    FSDataInputStream is = path.getFileSystem(new Configuration()).open(path);
    DataFileStream<GenericRecord> stream = new DataFileStream<>(is, new GenericDatumReader<>());
    Schema avroSchema = stream.getSchema();
    Record schema = AvroParser.parse(avroSchema);
    out.println("// Generated by magnolify-tools ");
    out.println("// Source: " + path);
    out.println();

    if (source) {
      out.println("/*");
      out.println(avroSchema.toString(true));
      out.println("*/");
      out.println();
    }

    out.println(SchemaPrinter.print(schema, width));
    return 0;
  }
}
